package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import com.gxlevi.util.HBaseUtil
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.api.scala._

case class ChannelBrowser(
                           var channelId: String,
                           var browser: String,
                           var date: String,
                           var pv: Long,
                           var uv: Long,
                           var newCount: Long,
                           var oldCount: Long
                         )

object ChannelBrowserTask extends BaseTask[ChannelBrowser] {
  override def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[ChannelBrowser] = {
    clickWideDataStream.flatMap {
      clickLog =>
        List(
          ChannelBrowser(
            clickLog.channelID,
            clickLog.browserType,
            clickLog.yearMonthDayHour,
            clickLog.count,
            clickLog.isHourNew,
            clickLog.isNew,
            isOld(clickLog.isNew, clickLog.isHourNew)
          ),
          ChannelBrowser(
            clickLog.channelID,
            clickLog.browserType,
            clickLog.yearMonthDay,
            clickLog.count,
            clickLog.isDayNew,
            clickLog.isNew,
            isOld(clickLog.isNew, clickLog.isDayNew)
          ),
          ChannelBrowser(
            clickLog.channelID,
            clickLog.browserType,
            clickLog.yearMonth,
            clickLog.count,
            clickLog.isMonthNew,
            clickLog.isNew,
            isOld(clickLog.isNew, clickLog.isMonthNew)
          )
        )
    }
  }

  override def keyBy(mapStream: DataStream[ChannelBrowser]): KeyedStream[ChannelBrowser, String] = {
    mapStream.keyBy {
      browser =>
        browser.channelId + ":" + browser.date + ":" + browser.browser
    }
  }

  override def reduce(windowedStream: WindowedStream[ChannelBrowser, String, TimeWindow]): DataStream[ChannelBrowser] = {
    windowedStream.reduce {
      (t1, t2) => {
        ChannelBrowser(t1.channelId, t1.browser, t1.date, t1.pv + t2.pv, t1.uv + t2.uv, t1.newCount + t2.newCount, t1.oldCount + t2.oldCount)
      }
    }
  }

  override def sinkToHBase(reduceDataStream: DataStream[ChannelBrowser]): Unit = {
    reduceDataStream.addSink {
      browser =>
        tableName = "channel_browser"
        rowKey = s"${browser.channelId}:${browser.browser}:${browser.date}"
        val browserColumn = "browser"

        val map = HBaseUtil.getMapData(tableName, rowKey, clfName, List(pvColumn, uvColumn, newCountColumn, oldCountColumn))

        HBaseUtil.putMapData(tableName, rowKey, clfName, Map(
          channelIdColumn -> browser.channelId,
          browserColumn -> browser.browser,
          dateColumn -> browser.date,
          pvColumn -> getTotal(map, pvColumn, browser.pv),
          uvColumn -> getTotal(map, uvColumn, browser.uv),
          newCountColumn -> getTotal(map, newCountColumn, browser.newCount),
          oldCountColumn -> getTotal(map, oldCountColumn, browser.oldCount)
        ))
    }
  }
}
